package org.qloudgen.borg.task;

import java.io.IOException;

import org.apache.commons.lang.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.qloudgen.borg.CommonValues;
import org.qloudgen.borg.beans.Cluster;
import org.qloudgen.borg.messaging.message.TaskMessage;
import org.qloudgen.borg.metadb.DefaultImpl;
import org.qloudgen.borg.utils.RemoteShellExec;

public class TaskExecutor {

	static Log log = LogFactory.getLog(TaskExecutor.class);
	
	private TaskMessage task;
	private RemoteShellExec exec;
	private Cluster cluster;
	private Thread taskLogWriter;
	
	public TaskExecutor(TaskMessage task){
		this.setTask(task);
	}
	
	private void init(){
		
		// get cluster info from metadb
		cluster = DefaultImpl.getClusterInfo(task.getClusterID());
		
		// new RemoteShellExec
		exec = new RemoteShellExec(cluster.getMasterIp(),cluster.getSysUser(),cluster.getSysUserPassword());
		try {
			exec.login();
		} catch (IOException e) {
			log.error(e.getMessage());
		}
		
		taskLogWriter = new Thread(){
			public void run() {
				String cmd = " tail -f " + CommonValues.getTaskLogFile(task.getCacheID());
				exec.exec(cmd, task);
            }
		};
	}
	
	public void start(){

		// init 
		init();
		
		if (task.isReload()) {
		// need to reload data and transform data 
			
			// execute
			if (!LoadData()) return;
			
			if (!TransformData()) return;
			
		}
		
		// process func 
		funcProcess();	
		
		this.exec.setRunnable(false);
		
		if (this.exec != null) this.exec.close();
	}
	
	public boolean LoadData(){
		
		if (task.getPreFailedStatus() == null 
			|| !task.getPreFailedStatus().equals(CommonValues.TASK_STATUS_LOADED)
			|| !task.getPreFailedStatus().equals(CommonValues.TASK_STATUS_TRANSFORMED)) {
			
			// load data 
			String shell = "sh " + cluster.getScriptPath() + "/" + CommonValues.getDSLoadFileName(task.getDsNo()) + " " + cluster.getDataPath();
			
			if (!execute(shell)) return false;
			
		}
		
		// update task status to loaded
		DefaultImpl.updateTaskStatus(task, CommonValues.TASK_STATUS_LOADED);
		
		return true;
	}
	
	public boolean TransformData(){
		
		if (task.getPreFailedStatus() == null 
			|| !task.getPreFailedStatus().equals(CommonValues.TASK_STATUS_TRANSFORMED)) {
			
			// transform data
			String shell = "sh " + cluster.getScriptPath() + "/" + CommonValues.getDSTransformFileName(task.getDsNo(),task.getFuncNo()) + " " + task.getDsNo() ;
			
			if (!execute(shell)) return false;
			
		}
		
		// update task status to transformed
		DefaultImpl.updateTaskStatus(task, CommonValues.TASK_STATUS_TRANSFORMED);
		
		return true;
	}

	public boolean funcProcess(){
		
		/* execute func process shell , DON NOT FORGET TO add cache_id to shell parameter, 
		 * that means every shell first parameter must be cache_id
		 */
		
		// add task log file
		if (!execute(" touch " + CommonValues.getTaskLogFile(task.getCacheID()))) return false;
		
		// start task log write thread 
		this.taskLogWriter.start();
		
		// execute func process
		String shell = "sh " + cluster.getScriptPath() + "/" + CommonValues.getFuncHandlerFileName(task.getFuncNo())
				+ " " + task.getCacheID() + " " + cluster.getJarPath() + " " + CommonValues.getTaskLogFile(task.getCacheID()) + " " + task.getDsNo();
		
		StringBuffer sb = new StringBuffer();
		if (StringUtils.isNotEmpty(task.getParams())){
			
			String[] splits = StringUtils.splitByWholeSeparator(task.getParams(),CommonValues.DEFAULT_FUNC_PARAM_DILIMTER);
			
			for( int i = 0; i< splits.length ; i++)
				sb.append(" " + splits[i]);
			
			shell += sb.toString();
		}
		
		if (!execute(shell)) return false;
		
		// update task status to success
		DefaultImpl.updateTaskStatus(task, CommonValues.TASK_STATUS_SUCCESS);
		
		// insert into cache
		DefaultImpl.insertTaskCache(this.task);
		
		return true;
	}
	
	private boolean execute(String shell){
		
		if (!((String)CommonValues.props.get("mode")).equalsIgnoreCase("test")) {
			if (!exec.exec2(shell)) {
				DefaultImpl.updateTaskStatus(task, CommonValues.TASK_STATUS_FAILED);
				return false;
			}
		} else {
			try {
				Thread.sleep(CommonValues.DEFAULT_TEST_SLEEP_TIME);
			} catch (InterruptedException e) {
				log.error(e.getMessage());
			}
		}
		
		return true;
	}

	public void killTask(){
		
		init();
		
		//kill process
		String shell = "ps -ef | grep '" + CommonValues.getFuncHandlerFileName(task.getFuncNo()) + " " + task.getCacheID() + "' | head  -1 | awk '{printf $2}' |xargs kill -s 9 ";
		exec.exec(shell);
			
		if (exec != null) exec.close();
	}
	
	/**
	 * @return the task
	 */
	public TaskMessage getTask() {
		return task;
	}

	/**
	 * @param task the task to set
	 */
	public void setTask(TaskMessage task) {
		this.task = task;
	}
	    
}
